package com.codejiwei.flink.table;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/8/8
 * desc: flink table api with execute
 **/
public class Flink_Table_API_execute {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                .column("col1", DataTypes.STRING())
                .column("col2", DataTypes.BIGINT())
                .build();

        tEnv.createTemporaryTable("SourceTable",
                TableDescriptor.forConnector("datagen")
                        .option(DataGenConnectorOptions.NUMBER_OF_ROWS, 2L)
                        .schema(schema)
                .build());

        tEnv.executeSql("create temporary table SinkTable with ('connector' = 'print') like SourceTable(excluding options)");
        tEnv.executeSql("create temporary table SinkTable2 with ('connector' = 'print') like SourceTable(excluding options)");

        //TODO execute with explicit sink
//        tEnv.from("SourceTable").executeInsert("SinkTable");

//        tEnv.executeSql("insert into SinkTable2 select * from SourceTable");

//        tEnv.createStatementSet()
//                .addInsertSql("insert into SinkTable select * from SourceTable")
//                .addInsertSql("insert into SinkTable2 select * from SourceTable")
//                .execute();

        //TODO execute with implicit local sink
//        tEnv.from("SourceTable").execute().print();
        tEnv.executeSql("select * from SourceTable ").print();

    }
}
